package com.jiayuan.cn.energy.test.ok;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkSqlStream {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(new Configuration());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        env.setParallelism(2);
//        EnvironmentSettings settings = EnvironmentSettings
//                .newInstance()
//                .inStreamingMode()
//                .build();
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 定义输入数据源
        String createSourceTableDdl = "CREATE TABLE csv_source (" +
                " user_id INT," +
                " product STRING," +
                " order_amount DOUBLE" +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'file:///path/input.csv'," +
                " 'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createSourceTableDdl);

        // 定义输出数据源
        String createSinkTableProduct = """
                CREATE TABLE test2 (
                        str STRING
                    ) WITH (
                        'connector'  = 'jdbc',
                        'url'        = 'jdbc:mysql://192.168.110.88:3306/helm?useUnicode=true&characterEncoding=utf-8&useSSL=false',
                        'table-name' = 'test2',
                        'driver'     = 'com.mysql.cj.jdbc.Driver',
                        'username'   = 'helm',
                        'password'   = 'db-999'
                    )
                """;

        tableEnv.executeSql(createSinkTableProduct);
        String query = "INSERT INTO test2   SELECT product as str  FROM csv_source ";
//        "SELECT user_id, product, sum(order_amount) as total_amount " +
//                +
//                "GROUP BY TUMBLE(proctime(), INTERVAL '20' second),user_id,product";
//                "GROUP BY user_id,product";
        tableEnv.executeSql(query);
//        env.execute("Flink SQL Job");
    }
}